package com.my;

import com.alibaba.fastjson.JSONObject;
import com.my.func.WithStateKeyedProcessFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

public class App {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStreamSource = env.socketTextStream("hadoop102", 7777);
        //异步乱序执行
        SingleOutputStreamOperator<JSONObject> source = AsyncDataStream.unorderedWait(dataStreamSource, new AsyncFunction<String, JSONObject>() {
            @Override
            public void asyncInvoke(String s, ResultFuture<JSONObject> resultFuture) throws Exception {
                resultFuture.complete(Collections.singleton(JSONObject.parseObject(s)));
            }
        }, 3, TimeUnit.SECONDS, 1000).disableChaining();
        KeyedStream<JSONObject, String> key = source.keyBy(jsonObject -> jsonObject.get("key").toString());
        key.process(new WithStateKeyedProcessFunction()).print(">>>>");
        env.execute();
    }


}
